package com.enjoy.rocketmq.quickstart;

import com.enjoy.rocketmq.Util;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

public class Consumer {
    public static void main(String[] args) throws Exception {
        Util.loadProperties();
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("pushConsumer");

        pushConsumer.setNamesrvAddr(System.getProperty("nameserver.addr"));
        pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        pushConsumer.subscribe("topic_a", "*");

        pushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            System.out.println(Thread.currentThread().getName() + "," + list);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        pushConsumer.start();
        System.out.println("Consumer start");
    }
}
